package com.intct.flink.window;

import com.intct.hbase.bean.WindowBean;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;

/**
 * @author gufg
 * @since 2025-06-27 14:26
 */
public class WindowReduceDemo {
    public static void main(String[] args) throws Exception {
        // 创建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 设置并行度
        env.setParallelism(1);

        // 从socket获取数据
        DataStreamSource<String> sourceDS = env.socketTextStream("jd-node", 8888);

        // Map转换(将字符转换为实体Bean)
        SingleOutputStreamOperator<WindowBean> mapDS = sourceDS.map(sources -> {
            String[] sourceList = sources.split(",");
            return WindowBean.builder()
                    .name(sourceList[0])
                    .ts(Long.valueOf(sourceList[1]))
                    .vc(Integer.valueOf(sourceList[2]))
                    .build();
        });

        // 按照指定Key分区
        KeyedStream<WindowBean, String> keyByDS = mapDS.keyBy(key -> key.getName());

        // 窗口-- 分配器(滚动窗口) 10秒一个窗口
        WindowedStream<WindowBean, String, TimeWindow> windowDS = keyByDS.window(TumblingProcessingTimeWindows.of(Time.seconds(30)));

        SingleOutputStreamOperator<WindowBean> reduceDS = windowDS.reduce(new ReduceFunction<WindowBean>() {
            @Override
            public WindowBean reduce(WindowBean value1, WindowBean value2) throws Exception {
                System.out.println("value1 : " + value1);
                System.out.println("value2 : " + value2);

                long currentTime = System.currentTimeMillis();
                String currentDate = DateFormatUtils.format(currentTime, "yyyy-MM-dd HH:mm:ss");

                return WindowBean.builder()
                        .name(value1.getName())
                        .ts(currentTime)
                        .vc(value1.getVc() + value2.getVc())
                        .build();
            }
        });

        reduceDS.print();

        env.execute();
    }
}
